v2/pkg/engine/datasource/graphql_datasource/graphql_datasource_test.go (1)

8905-9071: Consider adding test coverage for additional scenarios

The test suite for onSubscriptionStartFns is well-structured, but consider adding the following test cases for better coverage:

1. Test with multiple start functions to ensure they are all called in order
2. Test with nil/empty onSubscriptionStartFns to ensure no panics occur
3. Test the message emission scenario with the Start method (currently only AsyncStart is tested on lines 9053-9071)

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

Learnt from: SkArchon
PR: wundergraph/graphql-go-tools#1203
File: v2/pkg/engine/resolve/loader.go:63-67
Timestamp: 2025-07-02T15:28:02.122Z
Learning: In the graphql-go-tools codebase, result structs are consistently initialized with non-nil bytes.Buffer instances, making additional nil checks for res.out unnecessary defensive programming.

v2/pkg/engine/datasource/graphql_datasource/graphql_datasource_test.go (3)

8911-8914: Consider simplifying the channel struct for better readability.

The anonymous struct in the channel could be simplified since you're testing individual parameters separately.

Consider this approach for cleaner code:

-		startFnCalled := make(chan struct {
-			ctx   *resolve.Context
-			input []byte
-		}, 1)
+		var capturedCtx *resolve.Context
+		var capturedInput []byte
+		startFnCalled := make(chan struct{}, 1)

Then update the hook function:

				func(ctx *resolve.Context, input []byte) (bool, error) {
-					startFnCalled <- struct {
-						ctx   *resolve.Context
-						input []byte
-					}{ctx, input}
+					capturedCtx = ctx
+					capturedInput = input
+					startFnCalled <- struct{}{}
					return false, nil
				},

And the assertions:

-		called := <-startFnCalled
-		assert.Equal(t, ctx, called.ctx)
-		assert.Equal(t, []byte(`{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: \"#test\") { text createdBy } }"}`), called.input)
+		<-startFnCalled
+		assert.Equal(t, ctx, capturedCtx)
+		assert.Equal(t, []byte(`{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: \"#test\") { text createdBy } }"}`), capturedInput)

---

8927-8933: Add timeout safety to prevent hanging tests.

Consider adding a timeout to the channel receive operation to prevent tests from hanging if hooks fail to execute.

-		require.Len(t, startFnCalled, 1)
-		called := <-startFnCalled
+		require.Len(t, startFnCalled, 1)
+		select {
+		case called := <-startFnCalled:
+			// existing assertions
+		case <-time.After(time.Second):
+			t.Fatal("hook was not called within timeout")
+		}

---

8905-8953: Consider adding test cases for multiple hooks and close=true scenario.

The current tests provide good basic coverage, but consider adding these scenarios for more comprehensive testing:

1. Multiple hooks execution: Test that all hooks are called in order and verify behavior when some succeed and others fail.

2. Close=true scenario: Test the case where a hook returns (true, nil) to verify the close flag is properly propagated.

3. Mixed scenarios: Test combinations of multiple hooks with different return values.

Example additional test case:

t.Run("OnSubscriptionStart with multiple hooks", func(t *testing.T) {
    var callOrder []int
    subscriptionSource := SubscriptionSource{
        onSubscriptionStartFns: []OnSubscriptionStartFn{
            func(ctx *resolve.Context, input []byte) (bool, error) {
                callOrder = append(callOrder, 1)
                return false, nil
            },
            func(ctx *resolve.Context, input []byte) (bool, error) {
                callOrder = append(callOrder, 2)
                return false, nil
            },
        },
    }
    
    close, err := subscriptionSource.OnSubscriptionStart(ctx, []byte(`{}`))
    require.NoError(t, err)
    assert.False(t, close)
    assert.Equal(t, []int{1, 2}, callOrder)
})

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

v2/pkg/engine/datasource/graphql_datasource/configuration.go (1)

128-130: Document hook execution order and failure semantics.

Callers will benefit from knowing whether hooks run sequentially in declaration order, whether later hooks run if an earlier one returns an error, and whether they may run in parallel in the future.

Apply this diff:

-    // StartupHooks contains the method called when a subscription is started
-    StartupHooks []SubscriptionOnStartFn
+    // StartupHooks are invoked when a subscription is started.
+    // Execution: sequential, in declaration order. If any hook returns an error, remaining hooks are not executed and startup aborts.
+    // Hooks may call ctx.EmitSubscriptionUpdate to pre-inject messages before the subscription connection is established.
+    StartupHooks []SubscriptionOnStartFn

v2/pkg/engine/resolve/resolve_test.go (1)

4799-4800: Naming nit: consider a more descriptive return than “done”

Past feedback suggested renaming “done” to “accepted” for clarity. Here “done” seems to mean “the source has finished emitting.” Consider “finished” or “complete” to better communicate semantics.

v2/pkg/engine/resolve/context.go (2)

109-116: Provide a generic alias EmitEvent to align with prior discussion and future-proof naming.

Past discussion mentioned adding ctx.EmitEvent(...). Keeping EmitSubscriptionUpdate is fine, but a generic alias improves API ergonomics and avoids over-scoping to subscriptions.

No behavior change; just a thin wrapper:

// add in the same file (outside the changed lines)
func (c *Context) EmitEvent(data []byte) {
	c.EmitSubscriptionUpdate(data)
}

Optional: If you want to signal whether an emission happened, add a bool return in the alias (and leave the current method as-is):

func (c *Context) TryEmitEvent(data []byte) bool {
	emitEventFn := c.emitEventFn
	if emitEventFn == nil {
		return false
	}
	emitEventFn(data)
	return true
}

---

219-219: Avoid calling Free() concurrently with event emission.

Resetting emitEventFn to nil in Free() is correct, but it must not race with EmitSubscriptionUpdate. If that invariant is upheld by the engine, consider adding a brief note to Free() or the field comment to make this explicit. Otherwise, a sync primitive would be required here (not recommended unless you want to widen Context’s concurrency guarantees).

v2/pkg/engine/resolve/datasource.go (1)

30-40: Clarify when the hook runs (post-dedup) and what payload format Emit should use.

To make implementers’ lives easier, please tighten the doc comment to explicitly state that:

• The hook is invoked after request deduplication (mirroring Start’s contract).
• Hook code may call ctx.EmitSubscriptionUpdate with GraphQL subscription envelopes (data/errors per spec), and that returning an error aborts startup.

Apply this diff to the comment block:

-// HookableSubscriptionDataSource is a hookable interface for subscription data sources.
-// It is used to call a function when a subscription is started.
-// This is useful for data sources that need to do some work when a subscription is started,
-// e.g. to establish a connection to the data source or to emit updates to the client.
-// The function is called with the context and the input of the subscription.
-// The function is called before the subscription is started and can be used to emit updates to the client.
+// HookableSubscriptionDataSource enables a pre-start hook for subscriptions.
+// The hook is invoked after request deduplication but before the subscription Start/AsyncStart.
+// Implementations can perform startup work and/or emit updates via ctx.EmitSubscriptionUpdate.
+// Emitted payloads should be valid GraphQL subscription envelopes (data/errors as applicable).
+// Returning a non-nil error aborts subscription startup and is propagated to the client.
 type HookableSubscriptionDataSource interface {
 	// SubscriptionOnStart is called when a new subscription is created
 	// If an error is returned, the error is propagated to the client.
 	SubscriptionOnStart(ctx *Context, input []byte) (err error)
 }

v2/pkg/engine/datasource/graphql_datasource/configuration.go (2)

12-12: Watch cross-package coupling: config -> resolve dependency.

Importing resolve from the datasource configuration package tightens layering. It’s acceptable here, but if you want to keep configuration decoupled from engine internals, consider defining a minimal interface in this package (e.g., an EventEmitter with EmitEvent([]byte)) and type the hook against that instead of *resolve.Context. The backing engine can pass a compatible adapter.

---

107-111: Strengthen the hook contract in docs (immutability and abort behavior).

Make it explicit that the input must be treated as read-only and that returning an error prevents the subscription from starting.

Apply this diff to the comment:

-// SubscriptionOnStartFn defines a hook function that is called when a subscription starts.
-// It receives the resolve context and the input of the subscription.
-// The function can return an error.
+// SubscriptionOnStartFn defines a hook invoked just before the subscription starts.
+// It receives the resolve Context and the subscription input (read-only).
+// The hook may emit zero or more updates via ctx.EmitSubscriptionUpdate.
+// If the hook returns a non-nil error, startup is aborted and the error is propagated to the client.
 type SubscriptionOnStartFn func(ctx *resolve.Context, input []byte) (err error)

v2/pkg/engine/datasource/graphql_datasource/graphql_datasource_test.go (3)

4022-4022: Prefer a test-local HTTP client with a timeout over http.DefaultClient

Using http.DefaultClient can share global state and has no timeout by default, which risks test hangs. For determinism and isolation, use a local client with a finite timeout (also consistent with the pattern used in newSubscriptionSource further down).

Apply this diff:

-                       client: NewGraphQLSubscriptionClient(http.DefaultClient, http.DefaultClient, ctx),
+                       client: NewGraphQLSubscriptionClient(&http.Client{Timeout: 10 * time.Second}, http.DefaultClient, ctx),

---

8908-8910: Remove no-op defer; Done() doesn’t cancel context

defer ctx.Context().Done() is a no-op (Done() returns a channel; calling it doesn’t cancel anything). Either drop the line or create a cancellable parent context and pass it to resolve.NewContext if you intend to cancel.

Apply this diff to simply remove the no-op:

-        ctx := resolve.NewContext(context.Background())
-        defer ctx.Context().Done()
+        ctx := resolve.NewContext(context.Background())

---

8938-8954: Add ordering/short-circuit and empty-hooks tests for stronger guarantees

You already verify error propagation for a single hook. Two quick additions would lock in contract details:

• All hooks run sequentially and the chain short-circuits on the first error.
• No hooks (nil/empty slice) is a no-op and returns nil.

Apply this diff to extend the test function with two extra subtests:

   t.Run("SubscriptionOnStart calls subscriptionOnStartFns and returns error if one of the functions returns an error", func(t *testing.T) {
     ctx := resolve.NewContext(context.Background())
-    defer ctx.Context().Done()
+    defer ctx.Context().Done() // retained in this subtest for minimal change; you may remove as per earlier comment.

     subscriptionSource := SubscriptionSource{
       subscriptionOnStartFns: []SubscriptionOnStartFn{
         func(ctx *resolve.Context, input []byte) error {
           return errors.New("test error")
         },
       },
     }

     err := subscriptionSource.SubscriptionOnStart(ctx, []byte(`{"variables": {}, "extensions": {}, "operationName": "LiveMessages", "query": "subscription LiveMessages { messageAdded(roomName: \"#test\") { text createdBy } }"}`))
     require.Error(t, err)
     assert.ErrorContains(t, err, "test error")
   })
+
+  t.Run("SubscriptionOnStart executes hooks sequentially and short-circuits on error", func(t *testing.T) {
+    ctx := resolve.NewContext(context.Background())
+    // no-op defer removed as discussed earlier
+
+    secondCalled := make(chan struct{}, 1)
+    subscriptionSource := SubscriptionSource{
+      subscriptionOnStartFns: []SubscriptionOnStartFn{
+        func(ctx *resolve.Context, input []byte) error {
+          return errors.New("boom")
+        },
+        func(ctx *resolve.Context, input []byte) error {
+          secondCalled <- struct{}{}
+          return nil
+        },
+      },
+    }
+
+    err := subscriptionSource.SubscriptionOnStart(ctx, []byte(`{"variables":{}}`))
+    require.Error(t, err)
+    // Ensure the second hook did not run after the first failed.
+    select {
+    case <-secondCalled:
+      t.Fatal("second hook should not be called after an error in the first hook")
+    default:
+    }
+  })
+
+  t.Run("SubscriptionOnStart with no hooks is a no-op", func(t *testing.T) {
+    ctx := resolve.NewContext(context.Background())
+    subscriptionSource := SubscriptionSource{} // no hooks
+    err := subscriptionSource.SubscriptionOnStart(ctx, []byte(`{"variables":{}}`))
+    require.NoError(t, err)
+  })

v2/pkg/engine/datasource/graphql_datasource/graphql_datasource.go (2)

1946-1948: Document concurrency/immutability guarantees for these fields.

subscriptionOnStartFns is read-only after construction, which is a useful guarantee for thread-safety. A short comment here will prevent future accidental mutation.

 type SubscriptionSource struct {
-	client                 GraphQLSubscriptionClient
-	subscriptionOnStartFns []SubscriptionOnStartFn
+	client                 GraphQLSubscriptionClient
+	// subscriptionOnStartFns are configured at planning time and must be treated as immutable
+	// during subscription execution to avoid races.
+	subscriptionOnStartFns []SubscriptionOnStartFn
 }

---

1998-2006: Harden hook execution: recover from panics to avoid taking down the process.

Hooks are user/module-provided; a panic here would crash the server. Wrap execution with a recover() to convert panics into errors and fail the subscription gracefully.

-func (s *SubscriptionSource) SubscriptionOnStart(ctx *resolve.Context, input []byte) (err error) {
-	for _, fn := range s.subscriptionOnStartFns {
+func (s *SubscriptionSource) SubscriptionOnStart(ctx *resolve.Context, input []byte) (err error) {
+	// Convert panics in hooks into returned errors to protect the engine.
+	defer func() {
+		if r := recover(); r != nil {
+			err = fmt.Errorf("subscription on-start hook panic: %v", r)
+		}
+	}()
+	for _, fn := range s.subscriptionOnStartFns {
 		err = fn(ctx, input)
 		if err != nil {
 			return err
 		}
 	}
 	return
 }

Optionally add a brief godoc on the method clarifying: sequential execution, stop-on-first-error, and that any error prevents the subscription from starting.

v2/pkg/engine/resolve/resolve.go (2)

633-634: Return value ignored in this path; consider renaming or removing the bool.

go startHook() runs asynchronously and the return value is dropped. Either:

• rename to startHookAsync() and remove the bool return, or
• handle the result (e.g., by signaling back via an event) if you need to act on it.

Minor clean-up for clarity.

---

1189-1256: Avoid broadcasting hook updates — add a targeted subscription event

Verified: subscriptionUpdater.Update currently sends subscriptionEventKindTriggerUpdate (v2/pkg/engine/resolve/resolve.go:1196–1208) and the event loop (handleTriggerUpdate at v2/pkg/engine/resolve/resolve.go:785–811) fans that update out to every subscription under the trigger. The subscriptionEvent struct already includes id SubscriptionIdentifier (v2/pkg/engine/resolve/resolve.go:1258–1261), so the shape needed for a targeted event exists.

What to change (minimal, safe refactor):

• Add a new event kind next to the existing constants:
  • v2/pkg/engine/resolve/resolve.go (around 1276): add subscriptionEventKindTargetedUpdate.
• Add a targeted-emission API (one of these options):
  • Add a small method on subscriptionUpdater:
    • func (s *subscriptionUpdater) UpdateTo(id SubscriptionIdentifier, data []byte) { ... }
      — which sends subscriptionEvent{ triggerID: s.triggerID, id: id, kind: subscriptionEventKindTargetedUpdate, data: data }.
  • OR add Context.EmitSubscriptionUpdateToSelf(data) (filled by per-sub hook wiring) that emits a subscriptionEvent with kind subscriptionEventKindTargetedUpdate and the target id.
• Route targeted events in the single-threaded event loop (preserves ordering):
  • In the event switch (v2/pkg/engine/resolve/resolve.go: around handleEvent / handleTriggerUpdate), add handling for subscriptionEventKindTargetedUpdate that:
    • looks up the trigger (r.triggers[event.triggerID]),
    • looks up the specific subscription by event.id (SubscriptionIdentifier),
    • enqueues the same workItem{fn, false} for only that subscription’s worker (instead of iterating all subs).
  • Keep existing subscriptionEventKindTriggerUpdate behavior for backward compatibility.
• Tests:
  • Add a unit test under v2/pkg/engine/resolve/* (resolve_test.go area) that asserts EmitSubscriptionUpdateToSelf (or UpdateTo) sends the message only to the intended subscription and not to others. Update/extend existing hook tests (they already use ctx.EmitSubscriptionUpdate) to cover both broadcast and targeted cases.

Why this is safe and useful:

• subscriptionEvent already contains SubscriptionIdentifier, so no structural change is required.
• The event still flows through the single-threaded event loop (ordering preserved) but avoids per-sub filter invocation and accidental delivery to unrelated subscriptions.
• Keeps backward compatibility: existing broadcasts continue to work.

If you want I can:

• Produce the precise diff for these edits (const + new method + event-loop case + test stub).

v2/pkg/engine/resolve/resolve_test.go (7)

4790-4796: Test helper signature grew; consider options to avoid future churn

Adding the 4th parameter solved the immediate need, but it forced touching many callsites. To limit future churn if we add more toggles (e.g., more hooks), consider switching to functional options or a small options struct so callsites remain stable.

---

4812-4817: No-op when hook is nil is correct; add a short doc comment for call order

A tiny comment clarifying that this is invoked by the resolver before Start would aid future readers.

Apply this diff to add a brief comment:

+// SubscriptionOnStart is invoked by the resolver before the stream Start().
+// It lets tests simulate subscription startup hooks and optionally emit pre-start updates.
 func (f *_fakeStream) SubscriptionOnStart(ctx *Context, input []byte) (err error) {

---

5452-5484: Strengthen the hook invocation test by asserting the hook sees the expected input

You already assert the request in onStart. Also verify the exact input received by the hook closure to catch any wiring regressions.

Apply this diff inside the hook closure:

-        }, func(ctx *Context, input []byte) (err error) {
-            called <- true
-            return nil
-        })
+        }, func(ctx *Context, input []byte) (err error) {
+            expected := `{"method":"POST","url":"http://localhost:4000","body":{"query":"subscription { counter }"}}`
+            require.Equal(t, expected, string(input))
+            called <- true
+            return nil
+        })

---

5486-5515: Avoid flakiness: disable heartbeats when asserting exact message counts

This test asserts an exact count of 2 messages. Heartbeats are time-based and can introduce spurious messages under load. Disabling them here makes the test robust.

         ctx := &Context{
             ctx: context.Background(),
             ExecutionOptions: ExecutionOptions{
-                SendHeartbeat: true,
+                SendHeartbeat: false,
             },
         }

---

5517-5551: Same here: turn off heartbeats when asserting a precise number of updates

You assert workChanBufferSize+2 messages. Any heartbeat could break the equality. Turn them off for this test too.

         ctx := &Context{
             ctx: context.Background(),
             ExecutionOptions: ExecutionOptions{
-                SendHeartbeat: true,
+                SendHeartbeat: false,
             },
         }

---

5554-5623: Remove unused variables and simplify expectations in mixed-source test

• messagesDroppedFromHook is never incremented and always 0.
• hookCompleted is written to but never read; dead code and potential goroutine leak on channel writes if a future refactor blocks.

Cleaning these up reduces noise and makes the test intent clearer.

Apply this diff:

-        messagesDroppedFromHook := &atomic.Int32{}
         messagesToSendFromOtherSources := int32(100)
@@
-        firstMessageArrived := make(chan bool, 1)
-        hookCompleted := make(chan bool, 1)
+        firstMessageArrived := make(chan bool, 1)
@@
-            if counter == int(messagesToSendFromOtherSources)-1 {
-                select {
-                case hookCompleted <- true:
-                case <-time.After(defaultTimeout):
-                }
-            }
             return fmt.Sprintf(`{"data":{"counter":%d}}`, counter), counter == int(messagesToSendFromOtherSources)-1
@@
-            go func() {
+            go func() {
                 // Wait for the first message to arrive before sending updates
                 select {
                 case <-firstMessageArrived:
                     for i := 1; i < int(messagesToSendFromHook); i++ {
                         ctx.EmitSubscriptionUpdate([]byte(fmt.Sprintf(`{"data":{"counter":%d}}`, i+20000)))
                     }
-                    hookCompleted <- true
                 case <-time.After(defaultTimeout):
                     // if the first message did not arrive, do not send any updates
                     return
                 }
             }()
@@
-        assert.Equal(t, int32(messagesToSendFromHook+messagesToSendFromOtherSources-messagesDroppedFromHook.Load()+messagesHeartbeat), int32(len(recorder.Messages())))
+        assert.Equal(t, int32(messagesToSendFromHook+messagesToSendFromOtherSources+messagesHeartbeat), int32(len(recorder.Messages())))

---

5452-5673: Add a negative-path test: hook error prevents subscription start

We should cover the case where the hook returns an error: no upstream messages should flow, an error should be written once, and the subscription should complete/close. This guards against regressions in error propagation order.

Proposed test snippet:

t.Run("SubscriptionOnStart error aborts subscription before start", func(t *testing.T) {
    c, cancel := context.WithCancel(context.Background())
    defer cancel()

    fakeStream := createFakeStream(
        func(counter int) (string, bool) {
            // Should never be called if the hook aborts correctly
            return `{"data":{"counter":999}}`, true
        },
        1*time.Millisecond,
        func(input []byte) {
            assert.Equal(t, `{"method":"POST","url":"http://localhost:4000","body":{"query":"subscription { counter }"}}`, string(input))
        },
        func(ctx *Context, input []byte) error {
            return errors.New("hook failed")
        },
    )

    resolver, plan, recorder, id := setup(c, fakeStream)
    ctx := &Context{ctx: context.Background()}

    err := resolver.AsyncResolveGraphQLSubscription(ctx, plan, recorder, id)
    // Resolver should not fail starting the routine; error should be surfaced as a message
    assert.NoError(t, err)

    recorder.AwaitComplete(t, defaultTimeout)
    msgs := recorder.Messages()
    require.Len(t, msgs, 1)
    assert.JSONEq(t, `{"errors":[{"message":"hook failed"}],"data":null}`, msgs[0])
})

If you want, I can wire this in and adjust expectations based on your resolver’s exact error shape.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

v2/pkg/engine/resolve/context.go (1)

36-37: Document emitEventFn Concurrency Contract

The emitEventFn callback is only ever set once on each cloned Context—before invoking SubscriptionOnStart or Start—and thereafter is read by EmitSubscriptionUpdate. It must not be reassigned or cleared concurrently while events are being emitted (or while Free is called).

Please add this comment immediately above the field in v2/pkg/engine/resolve/context.go to codify the intended lifecycle:

--- a/v2/pkg/engine/resolve/context.go
+++ b/v2/pkg/engine/resolve/context.go
@@ type Context struct {
-   emitEventFn      func(data []byte)
+   // emitEventFn is the callback installed by the subscription engine
+   // to deliver updates to the client. It is assigned exactly once
+   // on each cloned Context—prior to calling SubscriptionOnStart or
+   // Start—and must not be mutated concurrently with
+   // EmitSubscriptionUpdate or Free.
+   emitEventFn      func(data []byte)

v2/pkg/engine/datasource/graphql_datasource/graphql_datasource_test.go (1)

8916-8936: Good: validates ctx and input are passed to hook

This subtest precisely checks both context wiring and input propagation to the hook. Clear and robust.

v2/pkg/engine/datasource/graphql_datasource/graphql_datasource.go (1)

453-455: LGTM: hooks are correctly wired into the SubscriptionSource instance.

Passing StartupHooks via subscriptionOnStartFns keeps configuration localized to the data source. Looks good.

v2/pkg/engine/resolve/resolve.go (3)

593-600: Good: detach and use a trigger-scoped cancelable context + updater.

Detaching from the request context and scoping cancellation to the trigger avoids coupling subscription lifetime to the HTTP request. The updater design cleanly feeds back into the event loop.

---

667-671: LGTM: synchronous gating of source start by the hook.

Starting the data source only if the hook succeeds prevents wasted connections and early events. Good sequencing.

---

1259-1265: Struct field reordering only; no functional impact detected.

Event struct remains compatible with existing switch handling. No concerns.

v2/pkg/engine/resolve/resolve_test.go (4)

4803-4810: Nice: injectable UniqueRequestID and onStart/hook fields improve test control

Allowing a custom uniqueRequestFn and the new hook field makes tests deterministic without coupling to internal hashing logic. LGTM.

---

4834-4837: Early exit for custom UniqueRequestID keeps tests deterministic

The override guard makes the hashing stable when tests need to force a specific key. Looks good.

---

5077-5082: Callsite update for createFakeStream hook param looks correct

Passing nil for the new subscriptionOnStartFn parameter keeps old behavior intact in tests that aren’t exercising hooks. No concerns here.

---

5625-5673: Two independent subscriptions to the same trigger verified end-to-end

Good coverage: exercising distinct connection/subscription IDs and the custom UniqueRequestID to validate isolation. LGTM.